--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit a1d2e30b42d30b570bd0082aece99fd9936e4892
Parents : 429702e
Author : Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-04-17T23:50:29-05:00
feat(translator): refactor Argos CLI detection and add integration tests for forwarding functionality
Changes
4 files changed, 195 insertions(+), 6 deletions(-)
Diff
diff --git a/meshchatx/src/backend/translator_handler.py b/meshchatx/src/backend/translator_handler.py
index fe07a6a3..2be09dc8 100644
--- a/meshchatx/src/backend/translator_handler.py
+++ b/meshchatx/src/backend/translator_handler.py
@@ -22,8 +22,17 @@ try:
except ImportError:
HAS_ARGOS_LIB = False
-HAS_ARGOS_CLI = shutil.which("argos-translate") is not None
-HAS_ARGOS = HAS_ARGOS_LIB or HAS_ARGOS_CLI
+ARGOS_CLI_EXECUTABLE_NAMES = ("argos-translate", "argostranslate")
+
+
+def _find_argos_cli_executable() -> str | None:
+ """Resolve the Argos CLI on PATH (checked at call time, not import time)."""
+ for name in ARGOS_CLI_EXECUTABLE_NAMES:
+ path = shutil.which(name)
+ if path:
+ return path
+ return None
+
LANGUAGE_CODE_TO_NAME = {
"en": "English",
@@ -83,9 +92,10 @@ class TranslatorHandler:
"LIBRETRANSLATE_URL",
"http://localhost:5000",
)
- self.has_argos = HAS_ARGOS
self.has_argos_lib = HAS_ARGOS_LIB
- self.has_argos_cli = HAS_ARGOS_CLI
+ self._argos_cli_executable = _find_argos_cli_executable()
+ self.has_argos_cli = self._argos_cli_executable is not None
+ self.has_argos = self.has_argos_lib or self.has_argos_cli
self.has_requests = HAS_AIOHTTP
async def _fetch_languages_async(self, url: str):
@@ -322,9 +332,12 @@ class TranslatorHandler:
msg = f"Invalid language codes: {source_lang} -> {target_lang}"
raise ValueError(msg)
- executable = shutil.which("argos-translate")
+ executable = self._argos_cli_executable or _find_argos_cli_executable()
if not executable:
- msg = "argos-translate executable not found in PATH"
+ msg = (
+ "Argos Translate CLI not found in PATH "
+ f"(tried: {', '.join(ARGOS_CLI_EXECUTABLE_NAMES)})"
+ )
raise RuntimeError(msg)
try:
diff --git a/tests/backend/test_lxmf_forwarding_integration.py b/tests/backend/test_lxmf_forwarding_integration.py
new file mode 100644
index 00000000..4b0aa17d
--- /dev/null
+++ b/tests/backend/test_lxmf_forwarding_integration.py
@@ -0,0 +1,125 @@
+# SPDX-License-Identifier: 0BSD
+
+"""Integration-style tests for LXMF handle_forwarding (rule path and reply path)."""
+
+import asyncio
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from meshchatx.meshchat import ReticulumMeshChat
+from meshchatx.src.backend.async_utils import AsyncUtils
+
+
+def _run_async_immediate(coro):
+ asyncio.run(coro)
+
+
+@pytest.fixture
+def forwarding_app():
+ app = MagicMock(spec=ReticulumMeshChat)
+ app.current_context = MagicMock()
+ app.current_context.database = MagicMock()
+ app.current_context.database.messages = MagicMock()
+ app.current_context.database.misc = MagicMock()
+ app.current_context.forwarding_manager = MagicMock()
+
+ app.send_message = AsyncMock()
+
+ app.handle_forwarding = ReticulumMeshChat.handle_forwarding.__get__(
+ app,
+ ReticulumMeshChat,
+ )
+ return app
+
+
+def _make_lxmf_message(source_hex: str, dest_hex: str):
+ msg = MagicMock()
+ msg.source_hash = bytes.fromhex(source_hex)
+ msg.destination_hash = bytes.fromhex(dest_hex)
+ msg.content = "body"
+ msg.title = "t"
+ msg.get_fields = MagicMock(return_value={})
+ return msg
+
+
+@pytest.mark.integration
+def test_handle_forwarding_forward_path_sends_via_alias(forwarding_app):
+ app = forwarding_app
+ ctx = app.current_context
+
+ src = "aa" * 16
+ dest_identity = "bb" * 16
+ forward_to = "cc" * 16
+ alias = "dd" * 16
+
+ ctx.database.messages.get_forwarding_mapping.return_value = None
+ ctx.database.misc.get_forwarding_rules.return_value = [
+ {
+ "source_filter_hash": None,
+ "forward_to_hash": forward_to,
+ },
+ ]
+ ctx.forwarding_manager.get_or_create_mapping.return_value = {
+ "alias_hash": alias,
+ }
+
+ msg = _make_lxmf_message(src, dest_identity)
+
+ with patch.object(AsyncUtils, "run_async", side_effect=_run_async_immediate):
+ app.handle_forwarding(msg, context=ctx)
+
+ app.send_message.assert_called_once()
+ kwargs = app.send_message.call_args[1]
+ assert kwargs["destination_hash"] == forward_to
+ assert kwargs["sender_identity_hash"] == alias
+ assert kwargs["content"] == "body"
+
+
+@pytest.mark.integration
+def test_handle_forwarding_reply_path_sends_to_original_sender(forwarding_app):
+ app = forwarding_app
+ ctx = app.current_context
+
+ original = "ee" * 16
+ alias = "ff" * 16
+
+ ctx.database.messages.get_forwarding_mapping.return_value = {
+ "original_sender_hash": original,
+ }
+ msg = _make_lxmf_message("aa" * 16, alias)
+
+ with patch.object(AsyncUtils, "run_async", side_effect=_run_async_immediate):
+ app.handle_forwarding(msg, context=ctx)
+
+ ctx.database.misc.get_forwarding_rules.assert_not_called()
+ app.send_message.assert_called_once()
+ kwargs = app.send_message.call_args[1]
+ assert kwargs["destination_hash"] == original
+ assert "sender_identity_hash" not in kwargs
+
+
+@pytest.mark.integration
+def test_handle_forwarding_skips_rule_when_source_filter_mismatches(forwarding_app):
+ app = forwarding_app
+ ctx = app.current_context
+
+ src = "aa" * 16
+ dest_identity = "bb" * 16
+ other = "99" * 16
+
+ ctx.database.messages.get_forwarding_mapping.return_value = None
+ ctx.database.misc.get_forwarding_rules.return_value = [
+ {
+ "source_filter_hash": other,
+ "forward_to_hash": "cc" * 16,
+ },
+ ]
+
+ msg = _make_lxmf_message(src, dest_identity)
+
+ with patch.object(AsyncUtils, "run_async", side_effect=_run_async_immediate):
+ app.handle_forwarding(msg, context=ctx)
+
+ app.send_message.assert_not_called()
+ ctx.forwarding_manager.get_or_create_mapping.assert_not_called()
diff --git a/tests/backend/test_translator_argos_integration.py b/tests/backend/test_translator_argos_integration.py
new file mode 100644
index 00000000..5f2daab4
--- /dev/null
+++ b/tests/backend/test_translator_argos_integration.py
@@ -0,0 +1,50 @@
+# SPDX-License-Identifier: 0BSD
+
+"""Optional integration checks for Argos CLI (skipped when not installed)."""
+
+import shutil
+
+import pytest
+
+from meshchatx.src.backend.translator_handler import (
+ ARGOS_CLI_EXECUTABLE_NAMES,
+ TranslatorHandler,
+ _find_argos_cli_executable,
+)
+
+
+def _argos_cli_on_path() -> bool:
+ return _find_argos_cli_executable() is not None
+
+
+@pytest.mark.integration
+def test_find_argos_cli_matches_shutil_which():
+ expected = None
+ for name in ARGOS_CLI_EXECUTABLE_NAMES:
+ expected = shutil.which(name)
+ if expected:
+ break
+ assert _find_argos_cli_executable() == expected
+
+
+@pytest.mark.integration
+@pytest.mark.skipif(not _argos_cli_on_path(), reason="Argos CLI not on PATH")
+def test_translate_en_es_via_cli_round_trip():
+ handler = TranslatorHandler(enabled=True)
+ assert handler.has_argos_cli
+ assert not handler.has_argos_lib
+
+ result = handler.translate_text("Hello", "en", "es", use_argos=True)
+ assert result["source"] == "argos"
+ assert result["source_lang"] == "en"
+ assert result["target_lang"] == "es"
+ assert len(result["translated_text"].strip()) > 0
+
+
+@pytest.mark.integration
+@pytest.mark.skipif(not _argos_cli_on_path(), reason="Argos CLI not on PATH")
+def test_get_supported_languages_includes_argos_when_libretranslate_down():
+ handler = TranslatorHandler(enabled=True)
+ langs = handler.get_supported_languages()
+ argos = [x for x in langs if x.get("source") == "argos"]
+ assert len(argos) >= 1
diff --git a/tests/backend/test_translator_handler_extended.py b/tests/backend/test_translator_handler_extended.py
index c0178bcb..e1557f2d 100644
--- a/tests/backend/test_translator_handler_extended.py
+++ b/tests/backend/test_translator_handler_extended.py
@@ -73,6 +73,7 @@ def test_translate_argos_cli(mock_run):
handler.has_argos_cli = True
handler.has_argos = True
handler.has_requests = False # Force CLI
+ handler._argos_cli_executable = "/usr/bin/argos-translate"
with patch("shutil.which", return_value="/usr/bin/argos-translate"):
result = handler.translate_text(
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────